package com.gx.mvp.rx;

import android.os.HandlerThread;
import io.reactivex.ObservableTransformer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.schedulers.Schedulers;
import java.util.concurrent.ExecutorService;

/**
 * rx线程调度工具类
 *
 * Created by Guoxin on 2019/12/05.
 */
public class RxSchedulers {

    private static volatile Scheduler ioScheduler;

    /**
     * 有边界的io线程池Scheduler
     *
     * 备注：rx自带的Schedules.io是无边界线程池，
     * 易导致栈内存泄露，所有这里条提供一个有边界线程池
     *
     * @return
     */
    public static Scheduler io() {
        if (ioScheduler == null) {
            synchronized (RxSchedulers.class) {
                if (ioScheduler == null) {
                    int maximumThreadSize = 80;
                    ExecutorService executor = CachedQueueExecutor.newCachedQueueThreadPool(maximumThreadSize);
                    //ExecutorService executor = new ThreadPoolExecutor(2, 20, 20L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(64), new ThreadPoolExecutor.DiscardOldestPolicy());
                    ioScheduler = Schedulers.from(executor);
                }
            }
        }
        return ioScheduler;
    }

    /**
     * 从io线程切换到android主线程
     *
     * @param <T>
     *
     * @return
     */
    public static <T> ObservableTransformer<T, T> io_main() {
        return observable -> observable.subscribeOn(Schedulers.io()).unsubscribeOn(io()).observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 创建Scheduler
     *
     * @param threadName
     *
     * @return
     */
    private static Scheduler createScheduler(String threadName) {
        HandlerThread handlerThread = new HandlerThread(threadName);
        handlerThread.start();
        return AndroidSchedulers.from(handlerThread.getLooper());
    }
}
